use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};

use model_graph_common::utils::new_uuid;
use model_graph_types::container::workflow::{
    WorkflowBlockChain, WorkflowBlockRef, WorkflowResult, WorkflowResultBuilder,
};
use model_graph_types::modeling::Value;
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use tokio::task::JoinHandle;

use crate::workflow::context::{Setter, WorkflowContext};
use crate::workflow::state_machine_manager::put_machine_result;

use super::state_machine_manager::{insert_machine, remove_machine};

#[derive(Debug, Serialize, Deserialize, Clone)]
pub enum WorkflowStateMachineState {
    Init,
    Running,
    Completed,
}

#[derive(Debug)]
pub struct WorkflowStateMachine {
    id: String,
    pub state: WorkflowStateMachineState,
    block_chain: WorkflowBlockChain,
    context: Arc<Mutex<WorkflowContext>>,
    running_queue: VecDeque<WorkflowBlockRef>,
    worker: Option<JoinHandle<()>>,
    sender: UnboundedSender<WorkflowBlockRef>,
    pub result: Option<WorkflowResult>,
}

impl WorkflowStateMachine {
    pub fn new(
        block_chain: WorkflowBlockChain,
        context: WorkflowContext,
    ) -> anyhow::Result<Arc<Mutex<WorkflowStateMachine>>> {
        let id = new_uuid()?;

        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel::<WorkflowBlockRef>();

        let sender_clone = sender.clone();

        let arc_context = Arc::new(Mutex::new(context));

        let id_temp = id.clone();
        let arc_context_temp = arc_context.clone();

        let machine = Arc::new(Mutex::new(WorkflowStateMachine {
            id: id_temp.clone(),
            block_chain,
            context: arc_context.clone(),
            state: WorkflowStateMachineState::Init,
            running_queue: VecDeque::new(),
            sender: sender_clone,
            result: None,
            worker: Some(tokio::task::spawn(async move {
                let mut builder = WorkflowResultBuilder::default();
                tracing::debug!("开始执行:{}", id_temp);
                loop {
                    //
                    if let Some(block) = receiver.recv().await {
                        let block_id = {
                            if let Ok(b) = block.clone().lock() {
                                b.id.clone()
                            } else {
                                String::new()
                            }
                        };
                        tracing::debug!("节点执行:{}", block_id);
                        // 执行
                        if let Ok(block_result) =
                            super::executor::execute(&block, arc_context_temp.clone()).await
                        {
                            //
                            let mut exit = false;

                            if block_result.is_return {
                                //是return
                                //返回结果
                                let mut outputs = HashMap::new();
                                for (k, v) in &block_result.result {
                                    if let Ok(sv) = v.try_into() {
                                        outputs.insert(k.clone(), sv);
                                    } else {
                                        tracing::error!("value转换失败!{:?}", k);
                                    }
                                }
                                builder.outputs(outputs);
                                exit = true;
                            } else {
                                //获取outputs，并激活
                                let outputs = {
                                    if let Ok(b) = block.clone().lock() {
                                        b.outputs.clone()
                                    } else {
                                        HashMap::new()
                                    }
                                };
                                for (key, value) in outputs {
                                    if let Some(source_handle) = &block_result.source_handle {
                                        //
                                        if key.contains(source_handle) {
                                            //检测节点是否激活
                                            let _ = sender.send(value.clone());
                                        }
                                    } else {
                                        tracing::error!("source_handle不包含:{},{:?}", key, value);
                                        //检测节点是否激活
                                        let _ = sender.send(value.clone());
                                    }
                                }
                            }
                            //将节点的输出，写入到context中
                            if let Err(err) =
                                arc_context_temp.set_node_result(&block_id, block_result.result)
                            {
                                tracing::error!("{}", err);
                            }
                            if exit {
                                break;
                            }
                        } else {
                            //调用执行失败
                        }
                    }
                }
                tracing::debug!("执行完成，设置结果");
                let result = if let Ok(result) = builder.build() {
                    result
                } else {
                    WorkflowResult::with_failed()
                };
                //保存结果， 设置状态
                let _ = put_machine_result(&id_temp, result);
            })),
        }));
        //插入该状态机
        insert_machine(id, machine.clone())?;

        Ok(machine)
    }

    pub fn new_and_start(
        block_chain: WorkflowBlockChain,
        context: WorkflowContext,
    ) -> anyhow::Result<Arc<Mutex<WorkflowStateMachine>>> {
        let arc_machine = WorkflowStateMachine::new(block_chain, context)?;
        {
            arc_machine
                .lock()
                .map_err(|err| anyhow::anyhow!("{}", err))?
                .start();
        }
        Ok(arc_machine)
    }
}

impl WorkflowStateMachine {
    /// 获取id
    pub fn id(&self) -> String {
        self.id.clone()
    }
    /// 获取状态
    pub fn state(&self) -> WorkflowStateMachineState {
        self.state.clone()
    }
    /// 获取状态
    pub fn set_state(&mut self, state: WorkflowStateMachineState) {
        self.state = state;
    }
    /// 启动
    pub fn start(&mut self) {
        //获取root
        let root = &self.block_chain.root;
        self.running_queue.push_back(root.clone());
        self.state = WorkflowStateMachineState::Running;
        let _ = self.sender.send(root.clone());
    }
    /// join
    pub async fn wait_result(&mut self) -> anyhow::Result<()> {
        let _ = self.worker.as_mut().unwrap().await?;
        //
        Ok(())
    }
    /// push 节点结果
    pub async fn push_node_result(
        &mut self,
        node_id: &String,
        result: HashMap<String, Value>,
    ) -> anyhow::Result<()> {
        //
        Ok(())
    }
}
